package com.bonc.wkafka.test;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;

import com.bonc.wkafka.utils.KafkaProps.ProcessingGuarantee;
import com.bonc.wkafka.utils.consumer.KafkaConsumerProps;
import com.bonc.wkafka.utils.consumer.WKafkaConsumer;
import com.bonc.wkafka.utils.consumer.translator.KafkaRecordTranslator;

public class ConsumerTest2 {
	public static void main(String[] args) {
		String groupid = "w1101";
		KafkaConsumerProps props = new KafkaConsumerProps("hadoop05:9092,hadoop06:9092,hadoop07:9092", new StringDeserializer(), new StringDeserializer(), "w2");
		props.setConsumerId(groupid)
		.setGroupId(groupid)
		.setPollIntervalMs(1000)
		.setProcessingGuarantee(ProcessingGuarantee.AT_LEAST_ONCE)
		.setRecordTranslator(new KafkaRecordTranslator<String, String>() {

			@Override
			public boolean handle(ConsumerRecord<String, String> record) {
				// TODO Auto-generated method stub
				System.out.println("partition: "+record.partition()+", value: "+record.value());
				return true;
			}
		});
//		.setFirstPollOffsetStrategy(FirstPollOffsetStrategy.EARLIEST);
		WKafkaConsumer<String, String> consumer = new WKafkaConsumer<String, String>(props);
		consumer.consume();
	}
}
